package com.swak.mongo;

import java.util.List;
import java.util.concurrent.CompletableFuture;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.util.Assert;

import com.mongodb.client.result.DeleteResult;
import com.mongodb.client.result.InsertManyResult;
import com.mongodb.client.result.InsertOneResult;
import com.mongodb.client.result.UpdateResult;
import com.mongodb.reactivestreams.client.MongoCollection;
import com.swak.entity.Page;
import com.swak.entity.Parameters;
import com.swak.mongo.json.Document;
import com.swak.mongo.json.Query;
import com.swak.mongo.json.Update;
import com.swak.utils.Lists;

/**
 * 客户端
 * 
 * @author lifeng
 */
public class MongoAsyncOptions implements DisposableBean {

	public String dbName;
	public com.mongodb.reactivestreams.client.MongoClient reactiveMongoClient;
	public com.mongodb.reactivestreams.client.MongoDatabase reactiveMongoDB;

	public MongoAsyncOptions(com.mongodb.reactivestreams.client.MongoClient reactiveMongoClient, String dbName) {
		this.reactiveMongoClient = reactiveMongoClient;
		this.reactiveMongoDB = reactiveMongoClient.getDatabase(dbName);
	}

	/**
	 * 根据ID获取数据
	 * 
	 * @param table
	 * @param id
	 * @return
	 */
	public CompletableFuture<Document> get(String table, Object id) {
		Assert.notNull(table, "table can not null");
		Assert.notNull(id, "id can not null");
		CompletableFuture<Document> future = new CompletableFuture<>();
		MongoCollection<Document> collection = this.reactiveMongoDB.getCollection(table, Document.class);
		Document query = new Document();
		query.put(Document.ID_FIELD, id);
		collection.find(query).first().subscribe(new Subscriber<Document>() {
			@Override
			public void onSubscribe(Subscription s) {
				s.request(1);
			}

			@Override
			public void onNext(Document t) {
				future.complete(t);
			}

			@Override
			public void onError(Throwable t) {
				future.completeExceptionally(t);
			}

			@Override
			public void onComplete() {

			}
		});
		return future;
	}

	/**
	 * 根据ID获取数据
	 * 
	 * @param table
	 * @param id
	 * @return
	 */
	public CompletableFuture<Document> get(String table, Query query) {
		Assert.notNull(table, "table can not null");
		Assert.notNull(query, "id can not null");
		CompletableFuture<Document> future = new CompletableFuture<>();
		MongoCollection<Document> collection = this.reactiveMongoDB.getCollection(table, Document.class);
		collection.find(query.getFilter()).first().subscribe(new Subscriber<Document>() {
			@Override
			public void onSubscribe(Subscription s) {
				s.request(1);
			}

			@Override
			public void onNext(Document t) {
				future.complete(t);
			}

			@Override
			public void onError(Throwable t) {
				future.completeExceptionally(t);
			}

			@Override
			public void onComplete() {

			}
		});
		;
		return future;
	}

	/**
	 * 根据ID获取数据
	 * 
	 * @param table
	 * @param id
	 * @return
	 */
	public CompletableFuture<Integer> count(String table, Query query) {
		Assert.notNull(table, "table can not null");
		Assert.notNull(query, "query can not null");
		CompletableFuture<Integer> future = new CompletableFuture<>();
		MongoCollection<Document> collection = this.reactiveMongoDB.getCollection(table, Document.class);
		collection.countDocuments(query.getFilter()).subscribe(new Subscriber<Long>() {
			@Override
			public void onSubscribe(Subscription s) {
				s.request(1);
			}

			@Override
			public void onNext(Long t) {
				future.complete(t.intValue());
			}

			@Override
			public void onError(Throwable t) {
				future.completeExceptionally(t);
			}

			@Override
			public void onComplete() {

			}
		});
		return future;
	}

	/**
	 * 根据ID获取数据
	 * 
	 * @param table
	 * @param id
	 * @return
	 */
	public CompletableFuture<Page<Document>> page(String table, Query query, Parameters param) {
		Assert.notNull(table, "table can not null");
		Assert.notNull(query, "query can not null");
		CompletableFuture<Page<Document>> future = new CompletableFuture<>();
		MongoCollection<Document> collection = this.reactiveMongoDB.getCollection(table, Document.class);
		collection.countDocuments(query.getFilter()).subscribe(new Subscriber<Long>() {

			@Override
			public void onSubscribe(Subscription s) {
				s.request(1);
			}

			@Override
			public void onNext(Long t) {
				param.setRecordCount(t.intValue());
				_query(collection, query, param, future);
			}

			@Override
			public void onError(Throwable t) {
				future.completeExceptionally(t);
			}

			@Override
			public void onComplete() {

			}
		});
		return future;
	}

	private void _query(MongoCollection<Document> collection, Query $query, Parameters param,
			CompletableFuture<Page<Document>> future) {
		collection.find(Document.class).filter($query.getFilter()).projection($query.getFields())
				.limit(param.getPageSize()).skip((param.getPageIndex() - 1) * param.getPageSize())
				.sort($query.getOrder()).subscribe(new Subscriber<Document>() {
					List<Document> results = Lists.newArrayList();

					@Override
					public void onSubscribe(Subscription s) {
						s.request(param.getPageSize());
					}

					@Override
					public void onNext(Document t) {
						results.add(t);
					}

					@Override
					public void onError(Throwable t) {
						future.completeExceptionally(t);
					}

					@Override
					public void onComplete() {
						future.complete(new Page<>(param, results));
					}
				});
	}

	/**
	 * 根据ID获取数据
	 * 
	 * @param table
	 * @param id
	 * @return
	 */
	public CompletableFuture<List<Document>> query(String table, Query query, int limit) {
		Assert.notNull(table, "table can not null");
		Assert.notNull(query, "query can not null");
		CompletableFuture<List<Document>> future = new CompletableFuture<>();
		MongoCollection<Document> collection = this.reactiveMongoDB.getCollection(table, Document.class);
		collection.find(Document.class).limit(limit).sort(query.getOrder()).filter(query.getFilter())
				.projection(query.getFields()).subscribe(new Subscriber<Document>() {
					List<Document> results = Lists.newArrayList();

					@Override
					public void onSubscribe(Subscription s) {
						s.request(limit);
					}

					@Override
					public void onNext(Document t) {
						results.add(t);
					}

					@Override
					public void onError(Throwable t) {
						future.completeExceptionally(t);
					}

					@Override
					public void onComplete() {
						future.complete(results);
					}
				});
		return future;
	}

	/**
	 * 插入数据
	 * 
	 * @param table
	 * @param docs
	 * @return
	 */
	public CompletableFuture<Void> insert(String table, Document... docs) {
		Assert.notNull(table, "table can not null");
		Assert.notNull(docs, "docs can not null");
		Assert.notEmpty(docs, "docs can not empty");
		CompletableFuture<Void> future = new CompletableFuture<>();
		MongoCollection<Document> collection = this.reactiveMongoDB.getCollection(table, Document.class);
		if (docs.length == 1) {
			collection.insertOne(docs[0]).subscribe(new Subscriber<InsertOneResult>() {
				@Override
				public void onSubscribe(Subscription s) {
					s.request(1);
				}

				@Override
				public void onNext(InsertOneResult t) {
					future.complete(null);
				}

				@Override
				public void onError(Throwable t) {
					future.completeExceptionally(t);
				}

				@Override
				public void onComplete() {

				}
			});
		} else {
			collection.insertMany(Lists.newArrayList(docs)).subscribe(new Subscriber<InsertManyResult>() {

				@Override
				public void onSubscribe(Subscription s) {
					s.request(1);
				}

				@Override
				public void onNext(InsertManyResult t) {
					future.complete(null);
				}

				@Override
				public void onError(Throwable t) {
					future.completeExceptionally(t);

				}

				@Override
				public void onComplete() {

				}

			});
		}
		return future;
	}

	/**
	 * 保存数据
	 * 
	 * @param table
	 * @param doc
	 * @return
	 */
	public CompletableFuture<Document> save(String table, Document doc) {
		Assert.notNull(table, "table can not null");
		Assert.notNull(doc, "doc can not null");
		CompletableFuture<Document> future = new CompletableFuture<>();
		MongoCollection<Document> collection = this.reactiveMongoDB.getCollection(table, Document.class);
		Object id = doc.get(Document.ID_FIELD);
		if (id == null) {
			collection.insertOne(doc).subscribe(new Subscriber<InsertOneResult>() {
				@Override
				public void onSubscribe(Subscription s) {
					s.request(1);
				}

				@Override
				public void onNext(InsertOneResult t) {
					future.complete(null);
				}

				@Override
				public void onError(Throwable t) {
					future.completeExceptionally(t);
				}

				@Override
				public void onComplete() {

				}
			});
		} else {
			com.mongodb.client.model.UpdateOptions options = new com.mongodb.client.model.UpdateOptions().upsert(true);
			Document filter = new Document();
			filter.put(Document.ID_FIELD, id);
			collection.updateOne(filter, new Update().set(doc), options).subscribe(new Subscriber<UpdateResult>() {

				@Override
				public void onSubscribe(Subscription s) {
					s.request(1);
				}

				@Override
				public void onNext(UpdateResult t) {
					future.complete(null);
				}

				@Override
				public void onError(Throwable t) {
					future.completeExceptionally(null);
				}

				@Override
				public void onComplete() {

				}
			});
			;
		}
		return future;
	}

	/**
	 * 修改数据
	 * 
	 * @param table
	 * @param doc
	 * @return
	 */
	public CompletableFuture<Long> update(String table, Document doc, Update update) {
		Assert.notNull(table, "table can not null");
		Assert.notNull(doc, "doc can not null");
		CompletableFuture<Long> future = new CompletableFuture<>();
		MongoCollection<Document> collection = this.reactiveMongoDB.getCollection(table, Document.class);
		com.mongodb.client.model.UpdateOptions options = new com.mongodb.client.model.UpdateOptions().upsert(true);
		Document filter = new Document();
		filter.put(Document.ID_FIELD, doc.get(Document.ID_FIELD));
		collection.updateOne(filter, update, options).subscribe(new Subscriber<UpdateResult>() {
			@Override
			public void onSubscribe(Subscription s) {
				s.request(1);
			}

			@Override
			public void onNext(UpdateResult t) {
				future.complete(t.getModifiedCount());
			}

			@Override
			public void onError(Throwable t) {
				future.completeExceptionally(t);
			}

			@Override
			public void onComplete() {
			}
		});
		return future;
	}

	/**
	 * 删除数据
	 * 
	 * @param table
	 * @param doc
	 * @return
	 */
	public CompletableFuture<Long> delete(String table, Document doc) {
		Assert.notNull(table, "table can not null");
		Assert.notNull(doc, "doc can not null");
		CompletableFuture<Long> future = new CompletableFuture<>();
		MongoCollection<Document> collection = this.reactiveMongoDB.getCollection(table, Document.class);
		Object id = doc.get(Document.ID_FIELD);
		if (id == null) {
			collection.deleteOne(doc).subscribe(new Subscriber<DeleteResult>() {

				@Override
				public void onSubscribe(Subscription s) {
					s.request(1);
				}

				@Override
				public void onNext(DeleteResult t) {
					future.complete(t.getDeletedCount());
				}

				@Override
				public void onError(Throwable t) {
					future.completeExceptionally(t);

				}

				@Override
				public void onComplete() {

				}
			});
			;
		} else {
			collection.deleteMany(doc).subscribe(new Subscriber<DeleteResult>() {

				@Override
				public void onSubscribe(Subscription s) {
					s.request(1);
				}

				@Override
				public void onNext(DeleteResult t) {
					future.complete(t.getDeletedCount());
				}

				@Override
				public void onError(Throwable t) {
					future.completeExceptionally(t);
				}

				@Override
				public void onComplete() {

				}
			});
			;
		}
		return future;
	}

	@Override
	public void destroy() throws Exception {
		try {
			this.reactiveMongoClient.close();
		} catch (Exception e) {
		}
	}
}
